package com.icreate.storm.bolt;

import com.icreate.storm.methods.common.StringMethod;
import com.icreate.storm.methods.indicators.fieldCreater.FieldCreater;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.LinkedList;
import java.util.List;
import java.util.Map;

/**
 * Created by hadoop on 17-4-5.
 */
public class XdrTypeBolt extends BaseRichBolt {
    //    private String PACK_SEPARATOR = "`";
    private String LINE_SEPARATOR = "\r\n";
    private Map<String, FieldCreater> xdrsMap;
    private OutputCollector collector;

    public XdrTypeBolt(Map<String, FieldCreater> xdrsMap) {
        this.xdrsMap = xdrsMap;
    }


    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        for (Map.Entry<String, FieldCreater> topic : this.xdrsMap.entrySet()) {
            declarer.declareStream(topic.getKey(), new Fields("fields"));
        }
    }

    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.collector = outputCollector;
    }

    @Override
    public void execute(Tuple tuple) {
        byte[] bytes = tuple.getBinaryByField("bytes");
        String str = new String(bytes);
//        collector.emit(, new Values(str));
        String type = StringMethod.getXdrTyp(str);
        if (!xdrsMap.containsKey(type))
            return;
        FieldCreater creater = this.xdrsMap.get(type);
        String[] strArray = str.split(LINE_SEPARATOR);
        List<Object> values = new LinkedList<>();
        for (int i = 0; i < strArray.length; i++) {
            Object value = creater.create(strArray[i]);
            if (value != null)
                values.add(value);
        }
        collector.emit(type, new Values(values));
        this.collector.ack(tuple);
    }

    private boolean isValidTopic(String topicName) {
        return this.xdrsMap.containsKey(topicName);
    }


}
